package com.etc;

import kafka.utils.ShutdownableThread;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Collections;
import java.util.Properties;

/**
 * @Author kalista
 * @Description kafka得消费者
 * @Date 2021/5/22  10:07
 **/
public class Consumer extends ShutdownableThread {

    public final static Logger log = LoggerFactory.getLogger(Consumer.class);

    private final KafkaConsumer<Integer, String> consumer;
    private final String topic;


    public Consumer(String topic, String groupId) {
        super(topic, false);
        Properties props = new Properties();

        props.put("bootstrap.servers", "hdp-01:9092,hdp-02:9092,hdp-03:9092");
        props.put("group.id", groupId);
        props.put("enable.auto.commit", "true");
        props.put("auto.offset.reset", "earliest");
        props.put("session.timeout.ms", "30000");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        consumer = new KafkaConsumer<>(props);
        this.topic = topic;
    }


    @Override
    public void doWork() {
        consumer.subscribe(Collections.singletonList(this.topic));
        ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofSeconds(1));
        System.out.println("消费到消息数:" + records.count());
        if (records.count() > 0) {
            for (ConsumerRecord<Integer, String> record : records) {
                log.warn("Received message: (" + record.key() + ", " + record.value() + ") at offset " + record.offset());
                String value = record.value();
            }
        }

    }

    @Override
    public String name() {
        return null;
    }

    @Override
    public boolean isInterruptible() {
        return false;
    }

    public static void main(String[] args) {
        Consumer consumer = new Consumer("my-topic","test001");
        consumer.start();
    }
}
